package com.atbeijing.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import javax.security.auth.callback.Callback;
import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class ConsumerManualAsync {

    public static void main(String[] args) {
        Properties conf = new Properties();
        try {
            conf.load(ConsumerManualAsync.class.getClassLoader().getResourceAsStream("consumer.properties"));
        } catch (IOException e) {
            e.printStackTrace();
        }
        conf.setProperty("group.id", "test03");
        conf.setProperty("enable.auto.commit", "false");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(conf);

        //消费指定topic
        consumer.subscribe(Collections.singleton("first"));
        //循环拉取
        while (true) {
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(10));
            //消费消息
            for (ConsumerRecord<String, String> consumerRecord : poll) {
                System.out.println(consumerRecord);
            }
            //手动异步提交offset,报告消费到哪里了
            consumer.commitAsync(new OffsetCommitCallback() {
                //回调函数
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                    System.out.println(offsets);
                }
            });
        }

        //关闭consumer
//            consumer.close();
    }

}
